package org.luxor.common.netty.jdk.nio.server;

import org.luxor.commons.core.utils.DateUtils;
import org.mockito.internal.util.io.IOUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Iterator;

public class MultiplexerTimeServerHandler implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(MultiplexerTimeServerHandler.class);

    private Selector selector;
    private ServerSocketChannel servChannel;
    private volatile boolean stop;

    /**
     * 初始化多路复用器，绑定监听端口
     */
    public MultiplexerTimeServerHandler(int port) {
        try {
            // 1、打开ServerSocketChannel，用于监听客户端的连接，它是所有客户端连接的父管道
            servChannel = ServerSocketChannel.open();
            // 2、绑定监听端口，设置连接为非阻塞模式
            servChannel.socket().bind(new InetSocketAddress(port), 1024);
            servChannel.configureBlocking(false);

            // 3、创建Reactor多路复用器
            selector = Selector.open();

            // 4、将用于接受客户端连接请求的通道(ServerSocketChannel)注册到多路复用器Selector上，监听连接事件(ACCEPT)
            servChannel.register(selector, SelectionKey.OP_ACCEPT);

            System.out.println("The time server is start in port : " + port);
        } catch (Exception e) {
            log.error("初始化多路复用器,绑定监听接口.失败!", e);
            System.exit(1);
        }
    }

    public void stop() {
        this.stop = true;
    }

    public void closeAll() {
        stop();
        // 多路复用器关闭后，所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭，所以不需要重复释放资源
        IOUtil.close(this.selector);
    }

    @Override
    public void run() {
        while (!stop) {
            try {
                // 设置多路复用器，检查通道就绪状态的最长等待时间(单位毫秒)
                selector.select(1000);

                // 5、无限循环处理多路复用器中,准备就绪的key-channel
                Iterator<SelectionKey> it = selector.selectedKeys().iterator();
                while (it.hasNext()) {
                    SelectionKey key = it.next();
                    try {
                        inputHandle(key);
                    } catch (Exception e) {
                        if (key != null) key.cancel();
                        if (key.channel() != null) key.channel().close();
                    } finally {
                        it.remove();
                    }
                }
            } catch (Exception e) {
                log.warn("多路复用器，循环处理准备就绪的key时，出现异常", e);
            }
        }
    }

    private void inputHandle(SelectionKey key) throws IOException {
        if (key.isValid()) {
            // 处理新接入的请求
            if (key.isAcceptable()) {
                // Accept the new connection
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                // 完成与客户端TCP三次握手,TCP物理链路通道正式建立
                SocketChannel sc = ssc.accept();
                // 设置异步非阻塞模式
                sc.configureBlocking(false);
                // 将客户端与服务器之间通信的网络通道(SocketChannel)注册到多路复用器Selector上，监听可读事件(READ)
                sc.register(selector, SelectionKey.OP_READ);
            }

            // 处理可读消息
            if (key.isReadable()) {
                SocketChannel sc = (SocketChannel) key.channel();
                // 读取客户端消息
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBuffer);
                if (readBytes > 0) {
                    readBuffer.flip();
                    byte[] bytes = new byte[readBuffer.remaining()];
                    readBuffer.get(bytes);
                    String body = new String(bytes, StandardCharsets.UTF_8);
                    System.out.println("The time server receive order:" + body);

                    // 写入响应消息( FIXME 存在'写半包'的问题 )
                    if ("QUERY TIME ORDER".equalsIgnoreCase(body)) {
                        doWrite(sc, DateUtils.format(new Date(), DateUtils.DATE_FORMAT_19_FORWARD_SLASH));
                    } else {
                        doWrite(sc, "BAD ORDER");
                    }
                }
                if (readBytes < 0) {
                    // 对端链路已关闭，需要关闭通道释放资源
                    key.cancel();
                    sc.close();
                }
            }
        }
    }

    private void doWrite(SocketChannel sc, String response) throws IOException {
        if (StringUtils.hasText(response)) {
            byte[] bytes = response.getBytes(StandardCharsets.UTF_8);
            ByteBuffer writeByte = ByteBuffer.allocate(bytes.length);
            writeByte.put(bytes);
            writeByte.flip();
            sc.write(writeByte);
        }
    }
}
